package avicit.bdp.dds.server.worker.task;

import avicit.bdp.common.dto.CalculateEngineConf;
import avicit.bdp.common.service.dto.TaskSpeedDTO;
import avicit.bdp.common.utils.ConfigUtils;
import avicit.bdp.common.utils.SpringApplicationContext;
import avicit.bdp.dds.dispatch.enums.ExecutionStatus;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dds.api.service.TaskLogService;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.common.thread.Stopper;
import avicit.bdp.dds.common.thread.ThreadUtils;
import avicit.bdp.dds.common.utils.LoggerUtils;
import avicit.bdp.dds.common.utils.OSUtils;
import avicit.bdp.dds.common.utils.YarnUtils;
import avicit.bdp.dds.server.entity.TaskExecutionContext;
import avicit.bdp.dds.server.utils.ProcessUtils;
import avicit.bdp.dds.server.worker.cache.TaskExecutionContextCacheManager;
import avicit.bdp.dds.server.worker.cache.impl.TaskExecutionContextCacheManagerImpl;
import avicit.platform6.commons.utils.ComUtil;
import org.apache.commons.lang3.StringUtils;
import org.jvnet.winp.WinProcess;
import org.slf4j.Logger;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static avicit.bdp.dds.common.Constants.EXIT_CODE_FAILURE;
import static avicit.bdp.dds.common.Constants.EXIT_CODE_SUCCESS;

/**
 * @author admin abstract command executor
 */
public abstract class AbstractCommandExecutor {
  /** rules for extracting application ID */
  protected static final Pattern APPLICATION_REGEX = Pattern.compile(Constants.APPLICATION_REGEX);

  /** process */
  private Process process;

  /** log handler */
  protected Consumer<List<String>> logHandler;

  /** logger */
  protected Logger logger;

  /** log list */
  protected final List<String> logBuffer;

  protected ArrayBlockingQueue<String> logSpeedBuffer;

  /** taskExecutionContext */
  protected TaskExecutionContext taskExecutionContext;

  protected TaskLogService taskLogService;

  private static final String TASK_SPEED_LOG_TAG = "[TaskSpeed]";
  private static final String TASK_SPEED_LOG_SPLIT_START = "@start@";
  private static final String TASK_SPEED_LOG_SPLIT_END = "@end@";

  /** taskExecutionContextCacheManager */
  private final TaskExecutionContextCacheManager taskExecutionContextCacheManager;

  private final ExecutorService handSpeedExecutorService;

  public AbstractCommandExecutor(
      Consumer<List<String>> logHandler, TaskExecutionContext taskExecutionContext, Logger logger) {
    this.logHandler = logHandler;
    this.taskExecutionContext = taskExecutionContext;
    this.logger = logger;
    this.logBuffer = Collections.synchronizedList(new ArrayList<>());
    logSpeedBuffer = new ArrayBlockingQueue<>(3000);
    this.taskExecutionContextCacheManager =
        SpringApplicationContext.getBean(TaskExecutionContextCacheManagerImpl.class);
    this.taskLogService = SpringApplicationContext.getBean(TaskLogService.class);
    String threadLoggerInfoName =
        String.format(
            LoggerUtils.TASK_LOGGER_SPEED_INFO_PREFIX + "-%s", taskExecutionContext.getTaskAppId());
    handSpeedExecutorService = ThreadUtils.newDaemonSingleThreadExecutor(threadLoggerInfoName);
  }

  /**
   * build process
   *
   * @param commandFile command file
   * @throws IOException IO Exception
   */
  private void buildProcess(String commandFile) throws IOException {
    // setting up user to run commands
    List<String> command = new LinkedList<>();

    if (OSUtils.isWindows()) {
      // setup command
      command.add(commandInterpreter());
      // command.add("/c");
      command.addAll(commandOptions());
      command.add(commandFile);

      ProcessBuilder builder = new ProcessBuilder();
      process =
          builder
              .directory(new File(taskExecutionContext.getExecutePath()))
              .command(command)
              .redirectErrorStream(true)
              .start();
    } else {
      // init process builder
      ProcessBuilder processBuilder = new ProcessBuilder();
      // setting up a working directory
      processBuilder.directory(new File(taskExecutionContext.getExecutePath()));
      // merge error information to standard output stream
      processBuilder.redirectErrorStream(true);

      // setting up user to run commands
      //            command.add("sudo");
      //            command.add("-i");
      //            command.add("-u");
      //            command.add(taskExecutionContext.getTenantCode());
      command.add(commandInterpreter());
      command.addAll(commandOptions());
      command.add(commandFile);

      // setting commands
      processBuilder.command(command);
      logger.info(">>>>>>>>command = {}", command);
      process = processBuilder.start();
    }

    // print command
    printCommand(command);
  }

  /**
   * task specific execution logic
   *
   * @param execCommand execCommand
   * @return CommandExecuteResult
   * @throws Exception if error throws Exception
   */
  public CommandExecuteResult run(String execCommand) throws Exception {

    CommandExecuteResult result = new CommandExecuteResult();

    if (StringUtils.isEmpty(execCommand)) {
      return result;
    }

    String commandFilePath = buildCommandFilePath();

    // create command file if not exists
    createCommandFileIfNotExists(execCommand, commandFilePath);

    // build process
    logger.info("buildProcess>>>>commandFilePath = {}", commandFilePath);
    buildProcess(commandFilePath);

    // parse process output
    parseProcessOutput(process);

    // 处理日志速度
    handSpeedBuffer();

    Integer processId = getProcessId(process);
    result.setProcessId(processId);

    // cache processId
    taskExecutionContext.setProcessId(processId);
    taskExecutionContextCacheManager.cacheTaskExecutionContext(taskExecutionContext);

    // print process id
    logger.info("process start, process id is: {}", processId);

    // if timeout occurs, exit directly
    long remainTime = getRemainTime();

    // waiting for the run to finish
    boolean status = process.waitFor(remainTime, TimeUnit.SECONDS);

    logger.info(
        "process has exited, execute path:{}, processId:{} ,exitStatusCode:{}",
        taskExecutionContext.getExecutePath(),
        processId,
        process.exitValue());

    // if SHELL task exit
    if (status) {
      // set appIds
      List<String> appIds = getAppIds(taskExecutionContext.getLogPath());
      result.setAppIds(String.join(Constants.COMMA, appIds));

      // SHELL task state
      result.setExitStatusCode(process.exitValue());

      // if yarn task , yarn state is final state
      if (process.exitValue() == 0) {
        result.setExitStatusCode(
            isSuccessOfYarnState(appIds) ? EXIT_CODE_SUCCESS : EXIT_CODE_FAILURE);
      }
    } else {
      logger.error(
          "process has failure , exitStatusCode : {} , ready to kill ...",
          result.getExitStatusCode());
      ProcessUtils.kill(taskExecutionContext);
      result.setExitStatusCode(EXIT_CODE_FAILURE);
    }

    return result;
  }

  /**
   * task specific execution logic
   *
   * @param execCommand execCommand
   * @return CommandExecuteResult
   * @throws Exception if error throws Exception
   */
  public CommandExecuteResult runCommand(List<String> execCommand) throws Exception {

    CommandExecuteResult result = new CommandExecuteResult();

    if (execCommand.isEmpty()) {
      return result;
    }

    // build process
    logger.info(">>>>build Process,command File Path = {}", StringUtils.join(execCommand, " "));
    // setting up user to run commands
    List<String> command = new LinkedList<>();

    if (OSUtils.isWindows()) {
      // setup command
      command.add(commandInterpreter());
      // command.add("/c");
      command.addAll(commandOptions());
      command.addAll(execCommand);

      ProcessBuilder builder = new ProcessBuilder();
      process =
          builder
              .directory(new File(taskExecutionContext.getExecutePath()))
              .command(command)
              .redirectErrorStream(true)
              .start();
    } else {
      // init process builder
      ProcessBuilder processBuilder = new ProcessBuilder();
      // setting up a working directory
      processBuilder.directory(new File(taskExecutionContext.getExecutePath()));
      // merge error information to standard output stream
      processBuilder.redirectErrorStream(true);

      // setting up user to run commands
      String tentant = ConfigUtils.getInstance().getString("tentant");
      if (StringUtils.isNotBlank(tentant)) {
        command.add("sudo");
        command.add("-u");
        command.add(tentant);
      }
      command.addAll(execCommand);

      // setting commands
      processBuilder.command(command);
      logger.info(">>>>>>>>command = {}", command);
      process = processBuilder.start();
    }

    // parse process output
    parseProcessOutput(process);

    // 处理日志速度
    handSpeedBuffer();

    Integer processId = getProcessId(process);
    result.setProcessId(processId);

    // cache processId
    taskExecutionContext.setProcessId(processId);
    taskExecutionContextCacheManager.cacheTaskExecutionContext(taskExecutionContext);

    // print process id
    logger.info("process start, process id is: {}", processId);

    // if timeout occurs, exit directly
    long remainTime = getRemainTime();

    // waiting for the run to finish
    boolean status = process.waitFor(remainTime, TimeUnit.SECONDS);

    logger.info(
        "process has exited, execute path:{}, processId:{} ,exitStatusCode:{}",
        taskExecutionContext.getExecutePath(),
        processId,
        process.exitValue());

    // if SHELL task exit
    if (status) {
      // set appIds
      List<String> appIds = getAppIds(taskExecutionContext.getLogPath());
      result.setAppIds(String.join(Constants.COMMA, appIds));

      // SHELL task state
      result.setExitStatusCode(process.exitValue());

      // if yarn task , yarn state is final state
      if (process.exitValue() == 0) {
        result.setExitStatusCode(
            isSuccessOfYarnState(appIds) ? EXIT_CODE_SUCCESS : EXIT_CODE_FAILURE);
      }
    } else {
      logger.error(
          "process has failure , exitStatusCode : {} , ready to kill ...",
          result.getExitStatusCode());
      ProcessUtils.kill(taskExecutionContext);
      result.setExitStatusCode(EXIT_CODE_FAILURE);
    }

    return result;
  }

  /**
   * cancel application
   *
   * @throws Exception exception
   */
  public void cancelApplication() throws Exception {
    if (process == null) {
      return;
    }

    // clear log
    clear();

    int processId = getProcessId(process);

    logger.info("cancel process: {}", processId);

    // kill , waiting for completion
    boolean killed = softKill(processId);

    if (!killed) {
      // hard kill
      hardKill(processId);

      // destory
      process.destroy();

      process = null;
    }
  }

  /**
   * soft kill
   *
   * @param processId process id
   * @return process is alive
   */
  private boolean softKill(int processId) {
    if (processId != 0 && process.isAlive()) {
      try {
        if (OSUtils.isWindows()) {
          ProcessUtils.killWindowsPorcess(processId);
        } else {
          // sudo -u user command to run command
          String cmd = String.format("sudo kill %s", processId);

          logger.info(
              "soft kill task:{}, process id:{}, cmd:{}",
              taskExecutionContext.getTaskAppId(),
              processId,
              cmd);

          Runtime.getRuntime().exec(cmd);
        }
      } catch (IOException e) {
        logger.info("kill attempt failed", e);
      }
    }

    return process.isAlive();
  }

  /**
   * hard kill
   *
   * @param processId process id
   */
  private void hardKill(int processId) {
    if (processId != 0 && process.isAlive()) {
      try {
        if (OSUtils.isWindows()) {
          ProcessUtils.killWindowsPorcess(processId);
        } else {
          String cmd = String.format("sudo kill -9 %d", processId);

          logger.info(
              "hard kill task:{}, process id:{}, cmd:{}",
              taskExecutionContext.getTaskAppId(),
              processId,
              cmd);

          Runtime.getRuntime().exec(cmd);
        }
      } catch (IOException e) {
        logger.error("kill attempt failed ", e);
      }
    }
  }

  /**
   * print command
   *
   * @param commands process builder
   */
  private void printCommand(List<String> commands) {
    try {
      String cmdStr = ProcessUtils.buildCommandStr(commands);
      logger.info("task run command:\n{}", cmdStr);
    } catch (IOException e) {
      logger.error(e.getMessage(), e);
    }
  }

  /** clear */
  private void clear() {
    if (!logBuffer.isEmpty()) {
      // log handle
      logHandler.accept(logBuffer);

      logBuffer.clear();
    }
    logSpeedBuffer.clear();
    // handSpeedExecutorService.shutdown();
  }

  /**
   * get the standard output of the process
   *
   * @param process process
   */
  private void parseProcessOutput(Process process) {
    String threadLoggerInfoName =
        String.format(
            LoggerUtils.TASK_LOGGER_THREAD_NAME + "-%s", taskExecutionContext.getTaskAppId());
    ExecutorService parseProcessOutputExecutorService =
        ThreadUtils.newDaemonSingleThreadExecutor(threadLoggerInfoName);
    parseProcessOutputExecutorService.submit(
        () -> {
          BufferedReader inReader = null;
          try {
            if (OSUtils.isWindows()) {
              inReader = new BufferedReader(new InputStreamReader(process.getInputStream(), "GBK"));
            } else {
              inReader =
                  new BufferedReader(new InputStreamReader(process.getInputStream(), "UTF-8"));
            }

            long lastFlushTime = System.currentTimeMillis();
            String line;
            while ((line = inReader.readLine()) != null) {
              // 从日志中获取纪录速度
              if (logSpeedBuffer != null
                  && line.contains(TASK_SPEED_LOG_TAG)
                  && line.contains(TASK_SPEED_LOG_SPLIT_START)) {
                logSpeedBuffer.offer(line);
              }
              logBuffer.add(line);
              lastFlushTime = flush(lastFlushTime);
            }
          } catch (Exception e) {
            logger.error(e.getMessage(), e);
          } finally {
            clear();
            close(inReader);
          }
        });
    parseProcessOutputExecutorService.shutdown();
  }

  private void handSpeedBuffer() {
    handSpeedExecutorService.submit(
        () -> {
          try {
            String taskInstanceId = taskExecutionContext.getTaskInstanceId();
            String processInstanceId = taskExecutionContext.getProcessInstanceId();
            while (logSpeedBuffer != null) {
              try {
                // 从日志中获取记录速度
                String speedStr = logSpeedBuffer.take();
                String taskSpeedStr =
                    speedStr.substring(
                        speedStr.lastIndexOf(TASK_SPEED_LOG_SPLIT_START) + 7,
                        speedStr.lastIndexOf(TASK_SPEED_LOG_SPLIT_END));
                TaskSpeedDTO taskSpeedDTO = JSONUtils.parseObject(taskSpeedStr, TaskSpeedDTO.class);
                if (taskSpeedDTO != null) {
                  // processInstanceId、taskInstanceId、endTimestamp相同则更新数据，不同则新增数据
                  TaskSpeedDTO taskSpeed =
                      taskLogService.selectOneTaskSpeed(
                          processInstanceId, taskInstanceId, taskSpeedDTO.getEndTimeStamp());
                  if (taskSpeed == null) {
                    taskLogService.insertTaskSpeed(
                        ComUtil.getId(),
                        processInstanceId,
                        taskInstanceId,
                        taskSpeedDTO.getStage(),
                        taskSpeedDTO.getTotalStage(),
                        taskSpeedDTO.getReadSucceedRecords(),
                        taskSpeedDTO.getReadSucceedBytes(),
                        taskSpeedDTO.getReadFailedRecords(),
                        taskSpeedDTO.getReadFailedBytes(),
                        taskSpeedDTO.getWriteSucceedRecords(),
                        taskSpeedDTO.getWriteSucceedBytes(),
                        taskSpeedDTO.getWriteFailedRecords(),
                        taskSpeedDTO.getWriteFailedBytes(),
                        taskSpeedDTO.getEndTimeStamp());
                  } else {
                    taskLogService.updateTaskSpeed(
                        taskSpeed.getId(),
                        processInstanceId,
                        taskInstanceId,
                        taskSpeedDTO.getStage(),
                        taskSpeedDTO.getTotalStage(),
                        taskSpeedDTO.getReadSucceedRecords(),
                        taskSpeedDTO.getReadSucceedBytes(),
                        taskSpeedDTO.getReadFailedRecords(),
                        taskSpeedDTO.getReadFailedBytes(),
                        taskSpeedDTO.getWriteSucceedRecords(),
                        taskSpeedDTO.getWriteSucceedBytes(),
                        taskSpeedDTO.getWriteFailedRecords(),
                        taskSpeedDTO.getWriteFailedBytes(),
                        taskSpeedDTO.getEndTimeStamp());
                  }
                }
              } catch (Exception e) {
                logger.error("insertTaskSpeed error===========================", e);
              }
            }
          } catch (Exception e) {
            e.printStackTrace();
          }
        });
  }

  /**
   * check yarn state
   *
   * @param appIds application id list
   * @return is success of yarn task state
   */
  public boolean isSuccessOfYarnState(List<String> appIds) {
    boolean result = true;
    try {
      for (String appId : appIds) {
        while (Stopper.isRunning()) {
          CalculateEngineConf conf = this.taskExecutionContext.getCalculateEngineConf();
          if (conf == null) {
            conf = new CalculateEngineConf();
          }

          ExecutionStatus applicationStatus = YarnUtils.getApplicationStatus(conf, appId);
          logger.info("appId:{}, final state:{}", appId, applicationStatus.name());
          if (applicationStatus.equals(ExecutionStatus.FAILURE)
              || applicationStatus.equals(ExecutionStatus.KILL)) {
            return false;
          }

          if (applicationStatus.equals(ExecutionStatus.SUCCESS)) {
            break;
          }
          Thread.sleep(Constants.SLEEP_TIME_MILLIS);
        }
      }
    } catch (Exception e) {
      logger.error(String.format("yarn applications: %s  status failed ", appIds.toString()), e);
      result = false;
    }
    return result;
  }

  public int getProcessId() {
    return getProcessId(process);
  }

  /**
   * get app links
   *
   * @param logPath log path
   * @return app id list
   */
  private List<String> getAppIds(String logPath) {
    List<String> logs = convertFile2List(logPath);

    List<String> appIds = new ArrayList<>();
    /** analysis log?get submited yarn application id */
    for (String log : logs) {
      String appId = findAppId(log);
      if (StringUtils.isNotEmpty(appId) && !appIds.contains(appId)) {
        logger.info("find app id: {}", appId);
        appIds.add(appId);
      }
    }
    return appIds;
  }

  /**
   * convert file to list
   *
   * @param filename file name
   * @return line list
   */
  private List<String> convertFile2List(String filename) {
    List<String> lineList = new ArrayList<>(100);
    File file = new File(filename);
    if (!file.exists()) {
      return lineList;
    }

    BufferedReader br = null;
    try {
      br =
          new BufferedReader(
              new InputStreamReader(new FileInputStream(filename), StandardCharsets.UTF_8));
      String line;
      while ((line = br.readLine()) != null) {
        lineList.add(line);
      }
    } catch (Exception e) {
      logger.error(String.format("read file: %s failed : ", filename), e);
    } finally {
      if (br != null) {
        try {
          br.close();
        } catch (IOException e) {
          logger.error(e.getMessage(), e);
        }
      }
    }
    return lineList;
  }

  /**
   * find app id
   *
   * @param line line
   * @return appid
   */
  private String findAppId(String line) {
    Matcher matcher = APPLICATION_REGEX.matcher(line);
    if (matcher.find()) {
      return matcher.group();
    }
    return null;
  }

  /**
   * get remain time?s?
   *
   * @return remain time
   */
  private long getRemainTime() {
    long usedTime =
        (System.currentTimeMillis() - taskExecutionContext.getStartTime().getTime()) / 1000;
    long remainTime = taskExecutionContext.getTaskTimeout() - usedTime;

    if (remainTime < 0) {
      throw new RuntimeException("task execution time out");
    }

    return remainTime;
  }

  /**
   * get process id
   *
   * @param process process
   * @return process id
   */
  private int getProcessId(Process process) {
    int processId = 0;

    try {
      Field f = process.getClass().getDeclaredField(Constants.PID);
      f.setAccessible(true);

      if (OSUtils.isWindows()) {
        // WinNT.HANDLE handle = (WinNT.HANDLE) f.get(process);
        // processId = Kernel32.INSTANCE.GetProcessId(handle);
        WinProcess proc = new WinProcess(process);
        processId = proc.getPid();
      } else {
        processId = f.getInt(process);
      }
    } catch (Throwable e) {
      logger.error(e.getMessage(), e);
    }

    return processId;
  }

  /**
   * when log buffer siz or flush time reach condition , then flush
   *
   * @param lastFlushTime last flush time
   * @return last flush time
   */
  private long flush(long lastFlushTime) {
    long now = System.currentTimeMillis();
    // when log buffer siz or flush time reach condition , then flush
    if (logBuffer.size() >= Constants.DEFAULT_LOG_ROWS_NUM
        || now - lastFlushTime > Constants.DEFAULT_LOG_FLUSH_INTERVAL) {
      lastFlushTime = now;
      // log handle
      logHandler.accept(logBuffer);

      logBuffer.clear();
    }
    return lastFlushTime;
  }

  /**
   * close buffer reader
   *
   * @param inReader in reader
   */
  private void close(BufferedReader inReader) {
    if (inReader != null) {
      try {
        inReader.close();
      } catch (IOException e) {
        logger.error(e.getMessage(), e);
      }
    }
  }

  protected List<String> commandOptions() {
    return Collections.emptyList();
  }

  protected abstract String buildCommandFilePath();

  protected abstract String commandInterpreter();

  protected abstract void createCommandFileIfNotExists(String execCommand, String commandFile)
      throws IOException;
}
